package org.study.hdfs;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.util.Iterator;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.junit.Before;
import org.junit.Test;

public class HDFSTest {

	@Before
	public void before() {
		System.setProperty("hadoop.home.dir", System.getProperty("user.dir") + "/src/main/resources");
	}

	// 将本地文件上传到hdfs
	@Test
	public void upload() throws IOException {
		String target = "hdfs://10.1.130.142:9300/test/input/LICENSE.txt";
		FileInputStream fis = new FileInputStream(new File(System.getProperty("user.dir") + "/src/main/resources/LICENSE"));// 读取本地文件
		Configuration config = new Configuration();
		FileSystem fs = FileSystem.get(URI.create(target), config);
		OutputStream os = fs.create(new Path(target));
		// copy
		IOUtils.copyBytes(fis, os, 4096, true);
		System.out.println("拷贝完成...");
	}

	@Test
	public void download() throws Exception {
		Configuration configuration = new Configuration();
		FileSystem fs = FileSystem.get(URI.create("hdfs://10.1.130.142:9300/test/out/result/part-00000"), configuration);
		InputStream in = fs.open(new Path(("hdfs://10.1.130.142:9300/test/out/result/part-00000")));
//		IOUtils.copyBytes(in, System.out, 4069, true);
		String localPath = System.getProperty("user.dir") + "/src/main/resources/result";
		OutputStream out = new FileOutputStream(localPath);
		IOUtils.copyBytes(in, out, 4096, true);
		fs.close();
	}

	@Test
	public void showFileInfo() throws Exception {
		final String pathString = "/";
		FileSystem fs = FileSystem.get(URI.create("hdfs://10.1.130.142:9300/"), new Configuration());
		final FileStatus[] listStatus = fs.listStatus(new Path(pathString));
		for (FileStatus fileStatus : listStatus) {
			final String type = fileStatus.isDir() ? "目录" : "文件";
			final short replication = fileStatus.getReplication();
			final String permission = fileStatus.getPermission().toString();
			final long len = fileStatus.getLen();
			final Path path = fileStatus.getPath();
			System.out.println(type + "\t" + permission + "\t" + replication + "\t" + len + "\t" + path);
		}
	}

	/**
	 * 获取配置的所有信息 首先，我们要知道配置文件是哪一个 然后我们将获取的配置文件用迭代器接收
	 * 实际上配置中是KV对，我们可以通过java中的Entry来接收
	 */
	public void showAllConf(Configuration conf) {
		Iterator<Map.Entry<String, String>> it = conf.iterator();
		while (it.hasNext()) {
			Map.Entry<String, String> entry = it.next();
			System.out.println(entry.getKey() + "=" + entry.getValue());
		}
	}
}
